package com.ficus.road.maintain.process;

import com.aliyun.oss.model.OSSObject;
import com.ficus.road.maintain.core.model.CarVideoRecord;
import com.ficus.road.maintain.core.model.Hit;
import com.ficus.road.maintain.core.model.ReplayVideoRequestDTO;
import com.ficus.road.maintain.core.model.SourceEnum;
import com.ficus.road.maintain.service.CarVideoRecordService;
import com.ficus.road.maintain.service.DataProcessorManager;
import com.ficus.road.maintain.service.HitService;
import com.ficus.road.maintain.util.AliyunOSSUtil;
import ficus.car_demo.FrameMeta;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.commons.lang3.tuple.Triple;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;

import java.io.IOException;
import java.io.InputStream;
import java.text.ParseException;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

import static com.ficus.road.maintain.util.AliyunOSSUtil.byteArrayToInt;
import static com.ficus.road.maintain.util.SsfFileUtil.sdf;

/**
 * @author
 */
@Slf4j
public class LoadSsfFileProcessor extends Thread {

    /**
     * 可播放的uuid放入playFlag
     * <uuid,<carId, timeStamp>>
     * 这里的AtomicReference好像没用, 并不会修改Map引用, 只是会修改Map里面的数据
     */
    @Getter
    public static volatile AtomicReference<Map<String, Pair<String, Long>>> playFlag = new AtomicReference<>(new ConcurrentHashMap<>());

    /**
     * 直播车
     * key: carId   value: sessionIds
     */
    @Getter
    public static volatile AtomicReference<Map<String, Set<String>>> liveCarSet = new AtomicReference<>(new ConcurrentHashMap<>());

    private final BlockingDeque<Triple<String, String, ReplayVideoRequestDTO>> blockingDeque;
    DataProcessorManager dataProcessorManager;
    AliyunOSSUtil aliyunOSSUtil;
    Executor taskAsyncPool;
    CarVideoRecordService carVideoRecordService;
    HitService hitService;

    public LoadSsfFileProcessor(String name, DataProcessorManager dataProcessorManager, AliyunOSSUtil aliyunOSSUtil, Executor taskAsyncPool, CarVideoRecordService carVideoRecordService, HitService hitService) {
        super(name);
        this.dataProcessorManager = dataProcessorManager;
        this.aliyunOSSUtil = aliyunOSSUtil;
        this.blockingDeque = new LinkedBlockingDeque<>(1);
        this.taskAsyncPool = taskAsyncPool;
        this.carVideoRecordService = carVideoRecordService;
        this.hitService = hitService;
    }

    @Override
    public void run() {
        while (true) {
            try {
                // 没有视频回放任务时, loadSsfFileProcessor阻塞
                // 且视频回放任务很快完成, 所以该线程绝大部分时间处于WAITING状态
                Triple<String, String, ReplayVideoRequestDTO> replayParams = blockingDeque.take();
                log.info("Video: LoadSsfFileProcessor take frame from deque");
                taskAsyncPool.execute(() -> readMetaFile(replayParams.getLeft(), replayParams.getMiddle(), replayParams.getRight()));
            } catch (Exception e) {
                log.info("解析io异常！");
            }
        }
    }

    /**
     * 播放历史视频的时候会调用这个方法
     * 一个播放历史视频的任务就调用一次put
     * carId, uuid, ReplayVideoRequestDTO
     *
     * @param triple
     */
    public void put(Triple<String, String, ReplayVideoRequestDTO> triple) {
        log.info("Video: LoadSsfFileProcessor's put, request: {}", triple);
        String uuid = triple.getMiddle();
        String carId = triple.getLeft();
        Map<String, Pair<String, Long>> playFlagMap = playFlag.get();
        playFlagMap.put(uuid, Pair.of(carId, System.currentTimeMillis()));
        try {
            this.blockingDeque.put(triple);
        } catch (InterruptedException e) {
            log.warn("insert triple param met interruption: ", e);
        }
    }

    /**
     * 实时播放
     *
     * @param carId
     * @param sessionId
     */
    public synchronized void liveReplay(String carId, String sessionId) {
        log.info("Video: liveReplay");
        Set<String> sessionSet = liveCarSet.get().getOrDefault(carId, new HashSet<>());
        if (sessionSet.isEmpty()) {
            new Thread(() -> readLiveMetaFile(carId), "liveReplayThread-" + carId).start();
        }
        //todo: 如果有车在直播
        sessionSet.add(sessionId);
        liveCarSet.get().put(carId, sessionSet);
    }

    /**
     * 关闭实时播放
     *
     * @param carId
     * @param sessionId
     */
    public synchronized void stopLiveReplay(String carId, String sessionId) {
        log.info("Video: stopLiveReplay");
        Set<String> sessionSet = liveCarSet.get().getOrDefault(carId, new HashSet<>());
        sessionSet.remove(sessionId);
        liveCarSet.get().put(carId, sessionSet);
    }

    /**
     * 关闭历史播放
     *
     * @param uuid
     */
    public void stopReplay(String uuid) {
        log.info("Video: stopReplay");
        playFlag.get().remove(uuid);
    }

    /**
     * 读直播数据
     *
     * @param carId
     */
    public void readLiveMetaFile(String carId) {
        log.info("Video: readLiveMetaFile");
        CarVideoRecord record = carVideoRecordService.findLastRecord(carId);
        while (record != null) {
            boolean compressed = record.isCompressed();
            CarVideoRecord nextRecord = carVideoRecordService.findNextRecord(record.getCarId(), record.getEndTime());
            String nextPicturePath;
            boolean nextCompressed;
            if (nextRecord != null) {
                nextPicturePath = aliyunOSSUtil.getOssUrl(nextRecord.getPicturePath());
                nextCompressed = nextRecord.isCompressed();
            } else {
                try {
                    // todo: 疑似死循环
                    // 如果没有下一个，直接继续等待
                    if (!liveCarSet.get().get(carId).isEmpty()) {
                        log.info("readLiveMetaFile has not next Record!");
                        Thread.sleep(50L);
                    }
                } catch (InterruptedException e) {
                    log.info("直播没找到下一个文件");
                }
                continue;
            }
            OSSObject ossObject = aliyunOSSUtil.getFile(nextRecord.getMetaPath());
            try {
                InputStream objectContent = ossObject.getObjectContent();
                while (true) {
                    //判断是否需要停止播放
                    if (liveCarSet.get().get(carId).isEmpty()) {
                        return;
                    }
                    byte[] lengthBytes = objectContent.readNBytes(4);
                    if (lengthBytes.length != 4) {
                        break;
                    }
                    //
                    int length = byteArrayToInt(lengthBytes);
                    byte[] bytes = objectContent.readNBytes(length);
                    if (bytes.length != length) {
                        break;
                    }
                    FrameMeta.FrameResult frameResult = FrameMeta.FrameResult.parseFrom(bytes);
                    dataProcessorManager.putLiveReplay(frameResult, aliyunOSSUtil.getOssUrl(nextRecord.getPicturePath()), nextPicturePath, compressed, nextCompressed, carId);
                }
            } catch (IOException e) {
                log.error("parse bytes met exception:", e);
            } finally {
                try {
                    ossObject.forcedClose();
                } catch (IOException e) {
                    log.error("close ossObject error:", e);
                }
            }
            record = nextRecord;
        }
    }

    public void readMetaFile(String carId, String uuid, ReplayVideoRequestDTO replayVideoRequestDTO) {
        log.info("Video: readMetaFile");
        String date = replayVideoRequestDTO.getDate();
        long timestampLong;
        Integer timestamp = replayVideoRequestDTO.getTimestamp();
        long time = 0;
        try {
            time = sdf.parse(date).getTime();
        } catch (ParseException e) {
            log.error("parse Date met exception:", e);
        }
        if (ObjectUtils.isEmpty(timestamp)) {
            timestampLong = time;
        } else {
            timestampLong = 1000L * timestamp;
        }
        // 找出近一天的数据
        List<CarVideoRecord> records = carVideoRecordService.findNextRecordByDate(carId, timestampLong, time + 24 * 60 * 60 * 1000);
        Criteria criteria = Criteria.where("first_hit_time").gte(timestampLong / 1000).lte(time / 1000 + 24 * 60 * 60).and("attrs.car_id").is(carId).and("attrs.source").is(SourceEnum.TAG.toString());
        Query query = new Query(criteria);
        List<Hit> hits = hitService.queryHits(query);
        Map<Long, List<Hit>> hitsMap = hits.stream().collect(Collectors.groupingBy(item -> {
            String[] split = item.getAttrs().getSceneImageName().split("/");
            String s = split[split.length - 1];
            String replace = s.replace(".jpg", "");
            if (StringUtils.isNumeric(replace)) {
                return Long.valueOf(replace);
            } else {
                return 0L;
            }
        }));
        if (CollectionUtils.isNotEmpty(hits)) {
            dataProcessorManager.putReplayHits(uuid, hitsMap, records.get(records.size() - 1).getEndTime() - records.get(0).getStartTime() + 10000);
        }
        for (int i = 0; i < records.size(); i++) {
            CarVideoRecord record = records.get(i);
            boolean compressed = record.isCompressed();
            CarVideoRecord nextRecord = null;
            if (i != records.size() - 1) {
                nextRecord = records.get(i + 1);
            }
            String nextPicturePath = "";
            boolean nextCompressed = false;
            if (nextRecord != null) {
                nextPicturePath = aliyunOSSUtil.getOssUrl(nextRecord.getPicturePath());
                nextCompressed = nextRecord.isCompressed();
            }
            OSSObject ossObject = aliyunOSSUtil.getFile(record.getMetaPath());
            try {
                InputStream objectContent = ossObject.getObjectContent();
                while (true) {
                    ReplayDataProcessor replayDataProcessor = dataProcessorManager.getReplayDataProcessorMap().get(uuid);
                    //判断是否需要停止播放(WebSocket关闭连接时, 会停止播放)
                    if (!playFlag.get().containsKey(uuid)) {
//                        ReplayDataProcessor replayDataProcessor = dataProcessorManager.getReplayDataProcessorMap().get(uuid);
                        if (replayDataProcessor != null) {
                            replayDataProcessor.stop();
                            dataProcessorManager.getReplayDataProcessorMap().remove(uuid);
                        }
                        return;
                    }  else if (replayDataProcessor!= null && !replayDataProcessor.isWSAvaliable()) {
                        log.info("检测到websocket连接超过指定时间仍断开,停止发送视频数据...");
                        replayDataProcessor.stop();
                        dataProcessorManager.getReplayDataProcessorMap().remove(uuid);
                        return;
                    }
                    byte[] lengthBytes = objectContent.readNBytes(4);
                    // 读取到文件最后一个片段后, 跳出循环
                    if (lengthBytes.length != 4) {
                        break;
                    }
                    int length = byteArrayToInt(lengthBytes);
                    byte[] bytes = objectContent.readNBytes(length);
                    if (bytes.length != length) {
                        break;
                    }
                    FrameMeta.FrameResult frameResult = FrameMeta.FrameResult.parseFrom(bytes);
                    //取出FrameResult，放入Replay
                    dataProcessorManager.putReplayFrame(frameResult, aliyunOSSUtil.getOssUrl(record.getPicturePath()), nextPicturePath, compressed, nextCompressed, uuid);
                }
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                try {
                    ossObject.forcedClose();
                } catch (IOException e) {
                    log.error("close ossObject error:", e);
                }
            }
        }
        dataProcessorManager.closeReplayWs(uuid);
    }

}
